// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/serializers/presto_header.h>

#include <pollux/serializers/presto_serializer_serialization_utils.h>

namespace kumo::pollux::serializer::presto::detail {
    /* static */
    Expected<PrestoHeader> PrestoHeader::read(
        ByteInputStream *source) {
        if (source->remainingSize() < kHeaderSize) {
            return melon::makeUnexpected(Status::Invalid(
                fmt::format("{} bytes for header", source->remainingSize())));
        }
        PrestoHeader header;
        header.numRows = source->read<int32_t>();
        header.pageCodecMarker = source->read<int8_t>();
        header.uncompressedSize = source->read<int32_t>();
        header.compressedSize = source->read<int32_t>();
        header.checksum = source->read<int64_t>();

        if (header.numRows < 0) {
            return melon::makeUnexpected(
                Status::Invalid(fmt::format("negative numRows: {}", header.numRows)));
        }
        if (header.uncompressedSize < 0) {
            return melon::makeUnexpected(Status::Invalid(
                fmt::format("negative uncompressedSize: {}", header.uncompressedSize)));
        }
        if (header.compressedSize < 0) {
            return melon::makeUnexpected(Status::Invalid(
                fmt::format("negative compressedSize: {}", header.compressedSize)));
        }

        return header;
    }

    /* static */
    std::optional<PrestoHeader> PrestoHeader::read(
        std::string_view *source) {
        if (source->size() < kHeaderSize) {
            return std::nullopt;
        }

        PrestoHeader header;
        header.numRows = readInt<int32_t>(source);
        header.pageCodecMarker = readInt<int8_t>(source);
        header.uncompressedSize = readInt<int32_t>(source);
        header.compressedSize = readInt<int32_t>(source);
        header.checksum = readInt<int64_t>(source);

        if (header.numRows < 0) {
            return std::nullopt;
        }
        if (header.uncompressedSize < 0) {
            return std::nullopt;
        }
        if (header.compressedSize < 0) {
            return std::nullopt;
        }

        return header;
    }
} // namespace kumo::pollux::serializer::presto::detail
